package com.lili.collection.mq.runner;

import com.lili.collection.core.config.OutputConfig;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.error.PluginRuntimeException;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.runner.stage.AbstractStageRunner;
import com.lili.collection.core.runner.StageRunner;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * 通用输入Runner
 *
 * @author lili
 * @date 2023/3/7 11:18
 */
@Slf4j
public class AsyncOutputRunner extends AbstractStageRunner {

    private OutputPlugin<OutputConfig> plugin;

    @Getter
    private PipeContext pipeContext;

    private RunnerResult outputRunnerResult;

    @Getter
    private RunnerMsgTransfer transfer;

    @Override
    public RunnerResult getRunnerResult() {
        return this.outputRunnerResult;
    }

    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer=transfer;
        this.plugin.setRunnerTransfer(transfer);
    }

    public AsyncOutputRunner(PipeContext pipeContext) {
        this.pipeContext = pipeContext;
        this.outputRunnerResult=new RunnerResult();
    }

    @Override
    public void preHandle() {
        if (this.plugin == null) {
            this.ensurePlugin();
        }
        this.plugin.preHandle();
    }

    private void ensurePlugin() {
        if (this.plugin != null) return;
        OutputConfig outputConfig = this.pipeContext.getContext().getOutputConfig();
        this.plugin = outputConfig.getPlugin();
        this.plugin.init(outputConfig);
        this.plugin.setRunnerResult(this.outputRunnerResult);
    }

    @Override
    public void run() {
        try {
            this.getTransfer().addThread(Thread.currentThread());
            ensurePlugin();
            while(true){
                if(this.transfer.isRunError()){
                    break;
                }
                checkStopSignal();
                StageContext stageContext = this.getPipeContext().getExchangeStageContext().take(StageRunner.StageType.ProcessGroup);
                this.plugin.execute(stageContext);
                triggerStageEvent(StageEvent.STAGE_CONTEXT_CHANGE, this.getStageType(), stageContext);
                if(stageContext.getStageContextState()== StageContext.StageContextState.COMPLETE){
                    break;
                }
            }
        } catch (Exception exception) {
            this.getTransfer().setRunError(true);
            this.getTransfer().interruptRunner();
            log.error("异步输出运行器异常!",exception);
        }finally {
            this.getTransfer().setOutputExit(true);
            if(Thread.currentThread().isInterrupted()){
                Thread.interrupted();
            }
            this.dispose();
        }
    }

    private void checkStopSignal() {
        if(this.pipeContext.getContext().getStopSignal()){
            throw new PluginRuntimeException("采集强制停止异常！");
        }
    }

    @Override
    public StageRunner.StageType getStageType() {
        return StageRunner.StageType.Output;
    }

    @Override
    public void dispose() {
        if (this.plugin != null)
            this.plugin.dispose();
    }

    @Override
    public void pause() {

    }

    @Override
    public void resume() {

    }

    @Override
    public void stop() {

    }

    @Override
    public int getDeleteOffsetNum() {
        return this.plugin.getDeleteOffsetNum();
    }

    @Override
    public String queryInitialOffset() {
        ensurePlugin();
        return this.plugin.queryInitialOffset();
    }
}
